
#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <fcntl.h>
#include <termios.h>
#include <sys/ioctl.h>
#include <sys/socket.h>
#include <netdb.h>
#include <netinet/in.h>
#include <sys/select.h>
#include <net/if.h>
#include <ctype.h>
#include "mqtt.h"
#include "posix_sockets.h"

int pty_master_open(char *salveName, size_t len)
{
	int msd;

	msd = posix_openpt(O_RDWR | O_NOCTTY);
	if (msd < 0) {
		perror("posix_openpt");
		exit(0);
	}

	if (grantpt(msd) < 0) {
		perror("grantpt");
		close(msd);
		exit(0);
	}

	if (unlockpt(msd) < 0) {
		perror("unlockpt");
		close(msd);
		exit(0);
	}

	ptsname_r(msd, salveName, len);
	return msd;
}

pid_t ptyFork(int *msd, char *slave_name, size_t len)
{
	int mfd = pty_master_open(slave_name, len);
	pid_t pid = fork();
	if (pid < 0) {
		perror("fork");
		exit(0);
	}

	if (pid) {
		*msd = mfd;
		return pid;
	}
	if (setsid() < 0) {
		perror("setsid");
		exit(0);
	}
	close(mfd);

	int sfd = open(slave_name, O_RDWR);
	if (sfd < 0) {
		perror("open");
		exit(0);
	}

	struct termios tty;
	tcgetattr(sfd, &tty);
	cfmakeraw(&tty);
	tcsetattr(sfd, TCSAFLUSH, &tty);

	dup2(sfd, STDIN_FILENO);
	dup2(sfd, STDOUT_FILENO);
	dup2(sfd, STDERR_FILENO);

	close(sfd);
	return 0;
}

int connect_serv(const char *ip, const char *service)
{
	struct addrinfo hints, *result;
	memset(&hints, 0, sizeof(hints));
	hints.ai_family = AF_UNSPEC;
	hints.ai_flags = AI_NUMERICHOST | AI_NUMERICSERV;
	hints.ai_socktype = SOCK_STREAM;
	hints.ai_socktype = 0;
	if (getaddrinfo(ip, service, &hints, &result) < 0) {
		perror("getaddrinfo");
		exit(0);
	}

	for (struct addrinfo *rp = result; rp; rp = rp->ai_next) {
		int fd = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol);
		if (fd < 0) {
			perror("socket");
			exit(0);
		}
		if (connect(fd, rp->ai_addr, rp->ai_addrlen) < 0) {
			perror("connect");
			close(fd);
			continue;
		}
		return fd;
	}
	return -1;
}

void *client_refresher(void *client)
{
	while (1) {
		mqtt_sync((struct mqtt_client *)client);
		usleep(100000U);
	}
	return NULL;
}
int msd = 0, len = 0;
void publish_callback(void **unused, struct mqtt_response_publish *published)
{
	/* note that published->topic_name is NOT null-terminated (here we'll change it to a c-string) */
	char *topic_name = (char *)malloc(published->topic_name_size + 1);
	memcpy(topic_name, published->topic_name, published->topic_name_size);
	topic_name[published->topic_name_size] = '\0';

	char buf[1024 * 1024 * 4];
	memcpy(buf, published->application_message, published->application_message_size);
	buf[published->application_message_size] = '\n';
	buf[published->application_message_size + 1] = 0;
	write(msd, buf, strlen(buf));

	free(topic_name);
}
uint8_t sendbuf[1024 * 1024]; /* sendbuf should be large enough to hold multiple whole mqtt messages */
uint8_t recvbuf[1024 * 1024]; /* recvbuf should be large enough any whole mqtt message expected to be received */
int connect_mqtt(struct mqtt_client *client, char *addr, char *port, char *sub_topic)
{
	int sockfd = open_nb_socket(addr, port);
	if (sockfd == -1) {
		perror("Failed to open socket: ");
		exit(0);
	}

	/* setup a client */

	mqtt_init(client, sockfd, sendbuf, sizeof(sendbuf), recvbuf, sizeof(recvbuf), publish_callback);
	/* Create an anonymous session */
	const char *client_id = NULL;
	/* Ensure we have a clean session */
	uint8_t connect_flags = MQTT_CONNECT_CLEAN_SESSION;
	/* Send connection request to the broker. */
	mqtt_connect(client, client_id, NULL, NULL, 0, NULL, NULL, connect_flags, 400);

	/* check that we don't have any errors */
	if (client->error != MQTT_OK) {
		fprintf(stderr, "error: %s\n", mqtt_error_str(client->error));
		exit(0);
	}

	/* start a thread to refresh the client (handle egress and ingree client traffic) */
	pthread_t client_daemon;
	if (pthread_create(&client_daemon, NULL, client_refresher, client)) {
		fprintf(stderr, "Failed to start client daemon.\n");
		exit(0);
	}

	/* subscribe */
	mqtt_subscribe(client, sub_topic, 0);

	/* start publishing the time */
	return 0;
}

int main()
{
	char slave_name[256], buf[1024];
	struct mqtt_client client;
	connect_mqtt(&client, "10.110.30.125", "1883", "testtopic/issue");

	pid_t pid = ptyFork(&msd, slave_name, sizeof(slave_name));
	if (pid == 0) {
		//execlp("telnet", "telnet", "192.168.141.1", NULL);
		execlp("bash", "bash", NULL);
		perror("execlp");
		exit(0);
	}

	fd_set rset, orset;
	FD_ZERO(&rset);
	FD_ZERO(&orset);
	FD_SET(msd, &orset);
	int maxfd = msd;
	for (;;) {
		rset = orset;
		int ret = select(maxfd + 1, &rset, NULL, NULL, NULL);
		if (ret < 0) {
			perror("select");
			exit(0);
		}

		if (FD_ISSET(msd, &rset)) {
			len = read(msd, buf, sizeof(buf));

			if (len < 0) {
				perror("read 2");
				exit(0);
			}
			buf[len] = 0;
			//fprintf(stderr, "msd=%s len=%d", buf, len);

			mqtt_publish(&client, "testtopic/post", buf, strlen(buf), MQTT_PUBLISH_QOS_0);
		}
	}
	return 0;
}